package com.audaque.springboot.foshanupload.redis.component;

import com.audaque.springboot.foshanupload.core.exception.RetryMaxException;
import com.audaque.springboot.foshanupload.core.exception.RetryingException;
import com.audaque.springboot.foshanupload.core.properties.ThreadProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import com.audaque.springboot.foshanupload.core.exception.RetryingException;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * @author zgb
 * @desc ...
 * @date 2022-02-10 17:17:46
 */

@Slf4j
@Component
public class AsyncSender {


    @Autowired
    private StringRedisTemplate stringRedisTemplate;



    @Recover
    public void recover(RetryingException e) throws RetryMaxException {
        String message = e.getMessage();
        String r = "重试次数已达上限:" + message;
        log.warn(r);
        //不要使用Exception，否则自定义异常信息丢失，要使用自定义 RetryMaxException extends RuntimeException
        throw new RetryMaxException(r);
    }

    //retry不能有返回值,retry代替while或递归，调用者不要catch RetryException否则会把Exception抓住，导致Exception丢失
    @Retryable(value = RetryingException.class, maxAttempts = Integer.MAX_VALUE, backoff = @Backoff(value =  5*1000, maxDelay = 1*60 * 1000, multiplier = 1.1))
    public void send( String name,String id, String productCountKey, String consumeCountKey,int availableProcessorsQueueSize,Integer waitConsumeInQueueSize) {
        String consumeCount = stringRedisTemplate.opsForValue().get(consumeCountKey);
        int consumeInteger = 0;
        if (StringUtils.isNotBlank(consumeCount)) {
            consumeInteger = Integer.parseInt(consumeCount);
        }

        String productCount = stringRedisTemplate.opsForValue().get(productCountKey);
        int productInteger = 0;
        if (StringUtils.isNotBlank(productCount)) {
            productInteger = Integer.parseInt(productCount);
        }


        Integer waitConsumeInQueue = productInteger - consumeInteger;
        log.debug(name + "消息生产计数-消费计数=队列中待消费计数=" + waitConsumeInQueue);

        Integer availableInQueue = availableProcessorsQueueSize - waitConsumeInQueue;
        log.debug(name + "队列长度-队列中待消费计数=队列剩余容量=" + availableInQueue);


        if (waitConsumeInQueueSize < waitConsumeInQueue) {
            String msg = "sendEvent：队列中待消费计数达到最大，触发重试";
            log.warn(msg);
            throw new RetryingException(msg);
        }
        if (availableInQueue < availableProcessorsQueueSize / 2) {
            String msg = "sendEvent：队列剩余容量小于一半，触发重试";
            log.warn(msg);
            throw new RetryingException(msg);
        }

        //发送异步事件，由异步线程处理
        //applicationContext.publishEvent(event);

        log.debug(name + "成功发送第" + productInteger + "条消息:id=" + id);

        log.debug("上报-开始发送-发送成功+1："+id);
        //获取指定key的值进行加1，如果value不是integer类型，会抛异常，如果key不存在会创建一个，默认value为0
        stringRedisTemplate.opsForValue().increment(productCountKey, 1L);


    }


}
